package com.futao.techsharingmq.kafka.basic.simple.interceptors;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author futao@gmail.com
 * @date 2021/11/4
 */
@Slf4j
public class ConsumerInterceptor implements org.apache.kafka.clients.consumer.ConsumerInterceptor<String, String> {

    AtomicInteger atomicInteger = new AtomicInteger(0);

    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
        StringBuilder sb = new StringBuilder();
        for (ConsumerRecord<String, String> record : records) {
            atomicInteger.getAndIncrement();
            sb.append(record.toString())
                    .append(System.lineSeparator());
        }
        log.info("记录消费者日志:{}", sb);
        log.info("-------------------------------------------------------{}", atomicInteger.get());
        return records;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        // log.info("提交偏移量之前被调用:{}", offsets);
    }

    @Override
    public void close() {
        System.out.println("-------------------------------------------------------" + atomicInteger.get());
        log.info("消费者拦截器被销毁");
    }

    @Override
    public void configure(Map<String, ?> configs) {
        log.info("消费者配置:{}", configs);
    }
}
